package com.sld.integrate.utils;

import com.sld.integrate.domain.RecDevMessage;
import com.sld.integrate.service.IRecDevCalculateService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 描述:
 *
 * @Author tanli
 * @create 2022-05-03 9:31
 */
@Slf4j
public class MessageQueueTaskManager {

    public static final String MSG_QUEUE_KEY = "list:calc:";

    // 任务管理
    Map<String, ScheduledFuture> taskManage = new HashMap<>();

    IRecDevCalculateService calculateService; //用于回调

    RedisTemplate redisTemplate;


    // 延迟任务执行线程池
    public ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(4);


    public MessageQueueTaskManager(RedisTemplate redisTemplate, IRecDevCalculateService calculateService) {
        this.redisTemplate = redisTemplate;
        this.calculateService = calculateService;
    }

    /**
     * 尝试计算，有可能是第一个数据，也有可能1.5分钟内正巧凑好7组，尝试计算就是这个意思。
     *
     * 第一个数据，必须激活清扫任务,激活一个1.5分钟的窗口期。
     *
     * @param msg
     * @return
     */
    public List<RecDevMessage> gatherMessage(RecDevMessage msg, int groupCount) {
        String devId = msg.getDevId();
        log.info("【凑数据】devId={}", devId);
        //来数据了，肯定要丢到队列里面去
        String key = MSG_QUEUE_KEY + devId;
        redisTemplate.opsForList().leftPush(key, msg);

        // 没有清扫任务，则激活一个delay任务，1.5分钟后，清空队列
        ScheduledFuture future = taskManage.get(devId);
        if (future == null) {
            ScheduledFuture<?> schedule = executorService.schedule(() -> {
                List list = redisTemplate.opsForList().range(key, 0, groupCount - 1);
                if (list != null && list.size() >=3 && list.size() <=6) {
                    calculateService.compensateCompute(devId, list);
                }
                reset(devId);
            }, 90, TimeUnit.SECONDS);
            taskManage.put(devId, schedule);
            log.info("【凑数据】devId={},无重置任务，重新激活一个", devId);
            redisTemplate.expire(key, 60, TimeUnit.MINUTES); // 任何key均需要过期，这是规范。
        }
        // 有任务缓存，说明窗口期还没结束。
        List list = redisTemplate.opsForList().range(key, 0, groupCount - 1);
        log.info("【凑数据】devId={},队列中数据size={}", devId, list.size());
        if (list.size() < groupCount) {
            return null;
        }
        log.info("【凑数据】devId={},凑够{}组了，下面需要重置", devId, groupCount);
        reset(devId);
        return list;
    }

    /**
     * 任务重置
     * @param devId
     */
    public void reset(String devId) {
        log.info("【队列重置】devId={}",devId);
        String key = MSG_QUEUE_KEY + devId;
        //二话不说，删除队列
        Boolean delete = redisTemplate.delete(key);
        log.info("【队列重置】删除队列res={}", delete);

        ScheduledFuture scheduledFuture = taskManage.get(devId);
        if (scheduledFuture != null) {
            boolean cancel = scheduledFuture.cancel(true);//停止该devId对应的延迟任务
            taskManage.remove(devId); // 删除map对应的信息
            log.info("【队列重置】删除任务slot cancel res={}", cancel);
        }
    }
}
